package com.xgzit.xgzmall.order.config;

import com.rabbitmq.client.AMQP;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * 指定 发消息的序列化转换器
 * 消息确认机制-可靠抵达
 *  事务消息-性能下降250倍
 *  引入消息确认
 * @Author xgz
 * @Date 2021-06-23 20:31
 */
@Slf4j
@Configuration
public class MyRabbitConfig {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }

    /** 定制 rabbitTemplate
     * 1、配置文件配置
     * 2、Broker 中 exchange 收到消息确认 设置 ConfirmCallback
     *          1、配置  publisher-confirm-type: correlated
     *          2、设置 ConfirmCallback
     *
     * 3、Broker 中 exchange 交给 queue 的回调 returnCallback
     *          1、配置  publisher-returns: true    template.mandatory: true
     *          2、设置 returnCallback
     *
     * 3、消费端确认 （保证每一个消息被正确消费，此时broker才可以删除这个消息）
     *  1、默认是自动确认的，只要消息收到，客户端就会自动确认，然后就可以接收下一个消息
     *      问题：
     *          收到很多消息，自动回复给服务器 ack，只有1个消息处理成功，然后宕机。就发生了消息丢失
     *      解决：可以手动确认。处理一个，确认一个。
     *          手动确认模式，只要没有明确发生ack，即使消费消息的服务器宕机，消息也不会丢失，消息会重新变成ready
     *          手动ack：加上配置：listener.simple.acknowledge-mode: manual ，再使用channel.basicAck()
     *  **/
    @PostConstruct  // MyRabbitConfig 对象创建完成以后，执行这个方法
    public void initRabbitTemplate(){

        // exchange 接收到消息回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback(){
            /**
            * 只要消息抵达Broker，就会调用。成功抵达 ack = true
            * @param correlationData: 当前消息的唯一关联数据（消息的唯一Id）
            * @param ack: 消息是否成功收到
            * @param cause:  失败原因
            * @return void
            */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                /**
                 * 1、做好消息确认机制【发消息者->交换机->队列->消费者】 全路径3个确认
                 * 2、每一个发送的消息都在数据库做好记录，定期扫描，重新发送  【补偿操作】
                 */
                // 服务器收到了； 修改消息的确认状态
                log.info("确认。。。correlationData: {}，ack: {}，cause: {}", correlationData, ack, cause);
            }
        });

        // exchange 投递到 queue 回调。 只要消息没有投递给指定的队列，就触发失败回调
        rabbitTemplate.setReturnCallback( new RabbitTemplate.ReturnCallback(){

            /**
            *
            * @param message: 投递失败的消息的详细信息
            * @param replyCode: 回复的状态码
            * @param replyText: 回复的文本
            * @param exchange: 投递这个消息的交换机
            * @param routingKey:  这个消息用的路由键
            * @return void
            */
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                // 报错误了 修改数据库当前消息的状态，变成 错误。
                log.info("确认。。。message: {}，replyCode: {}，replyText: {}, exchange:{},routingKey: {} ",
                        message, replyCode, replyText, exchange, routingKey);
            }
        });
    }
}
